package com.nami.dispatcher.domain.service.impl;

import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.core.util.RandomUtil;
import com.google.protobuf.ByteString;
import com.nami.dispatcher.domain.gateway.GrpMemberGateway;
import com.nami.dispatcher.domain.gateway.UserGateway;
import com.nami.dispatcher.domain.message.model.PacketTrans;
import com.nami.dispatcher.domain.middleware.mq.DomainEventPublisher;
import com.nami.dispatcher.domain.model.NtfGrpReceiverModel;
import com.nami.dispatcher.domain.qos.QosSendDaemonS2C;
import com.nami.dispatcher.domain.service.DispatchGrpDomainService;
import com.nami.gate.external.basic.bo.ConnectedUserBO;
import com.nami.gate.external.basic.cache.user.UserSessionRel;
import com.nami.gate.external.basic.constants.DownTopicConst;
import com.nami.im.network.model.enums.protocol.CMDEnum;
import com.nami.im.network.model.message.v2.Message;
import com.nami.transport.proto.codec.ImMessages;
import com.nami.transport.proto.codec.MqTransMsg;
import com.nami.transport.proto.codec.UpWsMsg;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.dromara.common.core.utils.SessionUtil;
import org.jetbrains.annotations.NotNull;
import org.springframework.stereotype.Component;

import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.function.Function;
import java.util.stream.Collectors;

import static com.nami.gate.external.basic.constants.CommonConst.COLON;

@Slf4j
@RequiredArgsConstructor
@Component
public class DispatchGrpDomainServiceImpl implements DispatchGrpDomainService {

    private final DomainEventPublisher mqSender;
    private final GrpMemberGateway grpMemberRepository;
    private final UserGateway userRepository;


    /**
     * 生成消息指纹
     *
     * @param userNo
     * @param clientMsgId
     * @return
     */
    @NotNull
    protected static String genUserMsgFingerChar(String userNo, String clientMsgId) {
        return userNo + COLON + clientMsgId;
    }

    @Override
    public void notifyReceiver(NtfGrpReceiverModel model) {
        forwardMsg2GrpMemberClients(model);
    }

    /**
     * 加上队列，就可以按照不同群成员等级，实现不同分发速度。比如单聊优先级高于群，vip群高于普通群，vip小群高与vip大群，
     */
    private void forwardMsg2GrpMemberClients(NtfGrpReceiverModel receiver) {
        ConnectedUserBO sender = receiver.getSender();
        Message message = receiver.getMessage();

        // TODO 拆分500人群，3000人群， 100000人群
        // date: 2024-12-15
        MqTransMsg.MessageHeader header = MqTransMsg.MessageHeader.newBuilder()
            .setCmd(ByteString.copyFrom(new byte[]{CMDEnum.PUBLISH.getCode()}))
            .setQos(ByteString.copyFrom(new byte[]{1}))
            .setVersion(ByteString.copyFrom(new byte[]{0}))
            .build();
        ImMessages.DownMsg downMsg = toDownMsg(message);
        Map<String, List<UserSessionRel>> map = receiver.getGrpClients();
        for (Map.Entry<String, List<UserSessionRel>> entry : map.entrySet()) {
            List<UserSessionRel> value = entry.getValue();
            if (CollectionUtil.isEmpty(value)) {
                continue;
            }
            // TODO 如果数据量大，此处需要拆分，进行小批量发送
            List<String> sessionIds = value.stream()
                .map(item -> SessionUtil.binding(item.getUserId(), item.getInstanceId()))
                .collect(Collectors.toList());
            List<MqTransMsg.UserPublishMsgCO> publishMsgs = value.stream()
                .map(item -> {
                    UpWsMsg.PublishMsg publishMsg = UpWsMsg.PublishMsg.newBuilder()
                        .setIndex(RandomUtil.randomInt())
                        .setTopic("msg")
                        .setTargetId(item.getUserId())
                        .setData(downMsg.toByteString())
                        .build();

                    MqTransMsg.UserPublishMsgCO userPublishMsg = MqTransMsg.UserPublishMsgCO.newBuilder()
                        .setPublish(publishMsg)
                        .setSessionId(SessionUtil.binding(item.getUserId(), item.getInstanceId()))
                        .build();
                    return userPublishMsg;
                }).collect(Collectors.toList());

            UserSessionRel userInfo = value.get(0);

            MqTransMsg.BatchDispatch2GatePublishResponse response = MqTransMsg.BatchDispatch2GatePublishResponse.newBuilder()
                .setMsgId(message.getMsgId())
                .addAllUserMsg(publishMsgs)
                .setMsgHeader(header)
                .setTimestamp(System.currentTimeMillis())
                .build();
            mqSender.publish(DownTopicConst.FLUSH_MORE_TOPIC, entry.getKey(), response.toByteArray());

            if (receiver.getQos() > 0) {
                Map<String, UserSessionRel> userDevicePairs = value.stream().distinct().collect(Collectors.toMap(key -> SessionUtil.binding(key.getUserId(), key.getPlatform()), Function.identity()));
                for (String sessionId : sessionIds) {
                    UserSessionRel l1User = userDevicePairs.get(sessionId);
                    if (null == l1User) {
                        continue;
                    }

                    UpWsMsg.PublishMsg publishMsg = UpWsMsg.PublishMsg.newBuilder()
                        .setIndex(RandomUtil.randomInt())
                        .setTopic("msg")
                        .setTargetId(SessionUtil.getUserId(sessionId))
                        .setData(downMsg.toByteString())
                        .build();
                    QosSendDaemonS2C.getInstance().put(genQosTransBo(publishMsg.toByteArray(), sessionId, message.getMsgId(),
                        l1User.getAppKey(), l1User.getConnectId()));
                }
            }

        }
        if (map.isEmpty()) {
            log.warn("[DISPATCHER-G] 在线群成员为空或为离线消息，找不到下发对象。senderId={}, recvId={}, msgId={}",
                sender.getUserId(), receiver.getTargetId(), message.getMsgId());
        }

    }

    private ImMessages.DownMsg toDownMsg(Message message) {
        ImMessages.DownMsg downMsg = ImMessages.DownMsg.newBuilder()
            .setClientUid(message.getClientUid())
            .setMsgId(message.getMsgId())
            .setMsgContent(ByteString.copyFromUtf8(message.getMsgContent()))
            .setMsgTime(message.getMsgTime())
            .setPlatform(message.getPlatform())
            .setChannelType(ImMessages.ChannelType.Group)
            .setSenderId(message.getSenderId())
            .setTargetId(message.getTargetId())
            .setMsgSeqNo(message.getMsgSeqNo())
            .setMsgType(message.getMsgType())
            .build();
        return downMsg;
    }

    @NotNull
    protected PacketTrans genQosTransBo(byte[] packet, String sessionId, String msgId,
                                        String appId, String mqTags) {
        PacketTrans packetTransBO = new PacketTrans();
        packetTransBO.setPacket(packet);
        packetTransBO.setQos(true);
        packetTransBO.setSessionId(sessionId);
        packetTransBO.setMessageId(msgId);
        packetTransBO.setAppKey(appId);

        packetTransBO.setMqTags(mqTags);

        packetTransBO.setIfGroupMsg(true);
        return packetTransBO;
    }

    @Override
    public Map<String, List<UserSessionRel>> findOnlineGrpClients(String appKey, List<String> memberUserIds) {

        if (CollUtil.isNotEmpty(memberUserIds)) {
            Map<String, List<UserSessionRel>> map = userRepository.getOnlineDevicesByMqTagsClassification(appKey,
                memberUserIds);
            return (map);

        }

        return Collections.EMPTY_MAP;
    }


}
